package com.yangzb;

import cn.hutool.core.io.FastByteArrayOutputStream;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.IoUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.io.*;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
import java.util.zip.GZIPOutputStream;
import java.util.zip.ZipOutputStream;

/**
 *
 * @author: yangzb
 * @date 2022/12/30 4:26 PM
 **/
@Slf4j
public class HdfsUtil {
    private static Configuration defaultConfiguration = new Configuration();

    static {
        defaultConfiguration.set("dfs.support.append", "true");
        defaultConfiguration.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
        defaultConfiguration.set("dfs.client.block.write.replace-datanode-on-failure.enable", "true");
        /**
         * 这个配置主要是针对打包的时候的问题：
         * hadoop filesystem相关的包有两个，分别是：hadoop-hdfs-2.7.1.jar和hadoop-common-2.7.1.jar
         * 指定即可，否则会出现java.io.IOException: No FileSystem for scheme: hdfs的异常
         */
        defaultConfiguration.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
    }

    public static Set<LocatedFileStatus> listFiles(String path, boolean recursive,Configuration configuration) throws IOException {
        return HdfsUtil.listFiles(new Path(path), true,configuration);
    }

    public static Set<LocatedFileStatus> listFiles(String path, boolean recursive) throws IOException {
        return HdfsUtil.listFiles(new Path(path), true,defaultConfiguration);
    }

    public static Set<LocatedFileStatus> listFiles(Path path, boolean recursive,Configuration configuration) throws IOException {
        try (FileSystem fileSystem = FileSystem.get(Optional.ofNullable(configuration).orElse(defaultConfiguration))){
            RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles(path, recursive);
            Set<LocatedFileStatus> files = new HashSet<>();
            while (locatedFileStatusRemoteIterator.hasNext()) {
                files.add(locatedFileStatusRemoteIterator.next());
            }
            return files;
        }
    }
    public static Set<LocatedFileStatus> listFiles(Path path, boolean recursive) throws IOException {
        return listFiles(path,recursive,defaultConfiguration);
    }

    public static String download(String hdfsFile, String localFile,FileSystem fileSystem) throws IOException {
        Path path = new Path(hdfsFile);
        FSDataInputStream inputStream = fileSystem.open(path);
        File file = FileUtil.writeFromStream(inputStream, localFile);
        IoUtil.close(inputStream);
        IoUtil.close(fileSystem);
        return file.getAbsolutePath();
    }

    public static String download(String hdfsFile, String localFile,Configuration configuration) throws IOException {
        return download(hdfsFile,localFile,FileSystem.get(Optional.ofNullable(configuration).orElse(defaultConfiguration)));
    }

    public static String download(String hdfsFile, String localFile) throws IOException {
        return download(hdfsFile,localFile,defaultConfiguration);
    }
}
